package cn.gxufe.reliable.message.admin.schedule.prodiucer;

import cn.gxufe.reliable.message.admin.domain.RMessage;
import cn.gxufe.reliable.message.admin.mapper.MessageMapper;
import cn.gxufe.reliable.message.admin.schedule.AbstractSchedule;
import cn.gxufe.reliable.message.admin.utils.PublishUtils;
import cn.gxufe.reliable.message.common.Message;
import cn.gxufe.reliable.message.common.ResultVo;
import com.netflix.hystrix.HystrixCommand;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;

import java.util.Date;
import java.util.List;

/**
 * 收到上游服务的 prepare 请求成功后，需要回调上游服务，确认confirm
 * @author yingjie.rong
 **/
@Component
public class TryConfirmService extends AbstractSchedule {


    private static final Logger logger = LoggerFactory.getLogger(TryConfirmService.class);

    private static final int MAX_TRY_CONFIRM = 5;

    @Autowired
    private MessageMapper messageMapper;


    @Scheduled(fixedDelay = 1000)
    public synchronized void exec(){
        List<RMessage> messages = messageMapper.selectCanConfirm(new Date(System.currentTimeMillis() - (3 * 1000)) );
        for( RMessage message: messages ) {
            tryConfirm(message);
        }
    }

    public void tryConfirm(RMessage message){
        MultiValueMap<String, Object> requestEntity = new LinkedMultiValueMap<>();
        requestEntity.add("messageId", message.getId());
        requestEntity.add("key", message.getKey());
        requestEntity.add("header", message.getHeader());
        HystrixCommand<ResultVo> hystrixCommand =  this.getHystrixCommand(message.getSourceService(),(t) -> {
            ResponseEntity<ResultVo> responseEntity = restTemplate.postForEntity("http://" + message.getSourceService() + CAN_CONFIRM_URL, requestEntity, ResultVo.class);
            return responseEntity.getBody();
        }, ResultVo.error(-1,"fallback"));
        ResultVo resultVo = hystrixCommand.execute();
        if( hystrixCommand.isSuccessfulExecution() ){
            logger.info("id = {}, key ={} call confirm successful !",message.getId(), message.getKey());
            if(Message.RESPONSE_CODE_OK.equals(resultVo.get(ResultVo.CODE))) {
                messageMapper.confirmUpdate(Message.STATUS_CONFIRM, message.getId(), Message.STATUS_PREPARE);
                message.setStatus( Message.STATUS_CONFIRM  );
                // 可选操作，会加入到内存队列，给另一个线程执行
                PublishUtils.publish(message);
            }else if(Message.RESPONSE_CODE_INCONSISTENT.equals(resultVo.get(ResultVo.CODE))) {
                /// 数据不一致
                messageMapper.confirmUpdate(Message.STATUS_INCONSISTENT, message.getId(), Message.STATUS_PREPARE);
            } else if(Message.RESPONSE_CODE_NOT_FOUNT.equals(resultVo.get(ResultVo.CODE))){
                if( message.getTryConfirm() >= MAX_TRY_CONFIRM ) {
                    messageMapper.cancelMessageForConfirm( Message.STATUS_CANCEL,message.getId(),Message.STATUS_PREPARE);
                }else{
                    messageMapper.confirmUpdate(Message.STATUS_PREPARE, message.getId(), Message.STATUS_PREPARE);
                }
            }
        } else  {
            logger.info("id = {}, key ={} call confirm error !",message.getId(), message.getKey());
            messageMapper.confirmUpdate(Message.STATUS_PREPARE, message.getId(), Message.STATUS_PREPARE);
        }
    }
}
